package com.semidata.trp.decision.business;

import io.netty.handler.codec.http.HttpResponseStatus;

import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReference;

import com.semidata.rtd.core.http.server.HttpRequestHandler;
import com.semidata.rtd.core.http.server.HttpResponse;
import com.semidata.rtd.core.http.server.HttpServiceInitializer;
import com.semidata.rtd.core.http.server.HttpSession;
import com.semidata.rtd.core.json.JsonObject;
import com.semidata.rtd.core.log.Logger;
import com.semidata.rtd.core.service.SendableService;
import com.semidata.trp.decision.job.DecisionJob;
import com.semidata.trp.decision.utils.ProtoUtils;
import com.semidata.trp.message.business.Business;
import com.semidata.trp.message.business.SyncReport;

public class BusinessService implements SendableService<BusinessEvent> {
	private LinkedBlockingQueue<BusinessEvent> queue = new LinkedBlockingQueue<>();
	private volatile boolean quit = false;
	private Thread thread = null;
	private AtomicReference<BusinessData> businessData = new AtomicReference<>(new BusinessData());
	private AtomicReference<BudgetData> budgetData = new AtomicReference<>(new BudgetData());
	private AtomicReference<ReportData> reportData = new AtomicReference<>(new ReportData());

	public BusinessService() {
	}

	@Override
	public void start() {
		thread = new Thread(new Runnable() {

			@Override
			public void run() {

				while (true) {
					try {

						BusinessEvent event = queue.poll(100,
						        TimeUnit.MILLISECONDS);
						if (event == null) {
							if (quit) {
								break;
							} else {
								continue;
							}
						}

						HttpSession session = event.getSession();
						try {

							// Logger.log("DEBUG", "" +
							// session.getBuffer().readableBytes());
							if (event.getType() == BusinessEvent.Type.BUSINESS) {
								

								// long start = System.currentTimeMillis();
								byte[] bytes = new byte[session.getBuffer()
								        .readableBytes()];
								session.getBuffer().readBytes(bytes);
								Business.BusinessData data = Business.BusinessData
								        .parseFrom(bytes);
								BusinessData newData = new BusinessData(data);
								businessData.set(newData);
								// Logger.log("DEBUG",
								// "parse business data time = " +
								// (System.currentTimeMillis() - start));
								// start = System.currentTimeMillis();
								// Logger.log("DEBUG",
								// "business data string length = " +
								// newData.toString().length());
								// Logger.log("DEBUG",
								// "parse business data to string, time = " +
								// (System.currentTimeMillis() - start));
							} else if (event.getType() == BusinessEvent.Type.BUDGET) {
								// long start = System.currentTimeMillis();
								byte[] bytes = new byte[session.getBuffer()
								        .readableBytes()];
								session.getBuffer().readBytes(bytes);
								Business.BudgetData data = Business.BudgetData
								        .parseFrom(bytes);
								BudgetData newData = new BudgetData(data);
								budgetData.set(newData);
								// Logger.log("DEBUG",
								// "parse budget data time = " +
								// (System.currentTimeMillis() - start));
								// Logger.log("DEBUG",
								// "budget data string length = " +
								// newData.toString().length());

							} else if (event.getType() == BusinessEvent.Type.REPORT){
								byte[] bytes = new byte[session.getBuffer()
								                        .readableBytes()];
								session.getBuffer().readBytes(bytes);
								SyncReport.SyncReportData data = SyncReport.SyncReportData.parseFrom(bytes);
								ReportData syncData = new ReportData(data);
								reportData.set(syncData);
								
							} else if (event.getType() == BusinessEvent.Type.BUSINESS_JSON) {
								JsonObject json = new JsonObject(session
								        .getBody());
								Business.BusinessData data = Business.BusinessData
								        .parseFrom(ProtoUtils
								                .businessDataJsonToProto(json));
								BusinessData newData = new BusinessData(data);
								businessData.set(newData);
							} else if (event.getType() == BusinessEvent.Type.BUDGET_JSON) {
								JsonObject json = new JsonObject(session
								        .getBody());
								Business.BudgetData data = Business.BudgetData
								        .parseFrom(ProtoUtils
								                .budgetDataJsonToProto(json));
								BudgetData newData = new BudgetData(data);
								budgetData.set(newData);
							} else {
								throw new Exception("not supported event type");
							}
							session.getResponse().end();
						} catch (Exception e) {
							e.printStackTrace();
							Logger.log("ERROR", e);
							session.getResponse().error(
							        HttpResponseStatus.BAD_REQUEST);
						}
					} catch (Exception e) {
					}
				}
			}
		});
		thread.start();
	}

	@Override
	public void stop() {
		quit = true;
		try {
			if (thread != null) {
				thread.join();
			}
		} catch (Exception e) {
		}
	}

	@Override
	public void send(BusinessEvent event) {
		try {
			queue.put(event);
		} catch (Exception e) {
		}
	}

	public BusinessData getBusinessData() {
		return businessData.get();
	}

	public BudgetData getBudgetData() {
		return budgetData.get();
	}
	
	public ReportData getReportData() {
		return reportData.get();
	}
}
